package com.gzx.qc.modular.voice.text.service.impl;

import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.gzx.qc.api.AiScannerApi;
import com.gzx.qc.core.config.QaKeywordInitializationConfig;
import com.gzx.qc.modular.voice.check.entity.GzxVoiceCheck;
import com.gzx.qc.modular.voice.check.service.GzxVoiceCheckService;
import com.gzx.qc.modular.voice.text.entity.GzxHtVoiceText;
import com.gzx.qc.modular.voice.text.service.GzxHtVoiceTextJobService;
import com.gzx.qc.modular.voice.text.service.GzxHtVoiceTextService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author xbl
 */
@Slf4j
@DS("qc")
@Service
@RequiredArgsConstructor
public class GzxHtVoiceTextServiceJobImpl implements GzxHtVoiceTextJobService {

    private final GzxVoiceCheckService gzxVoiceCheckService;

    private final GzxHtVoiceTextService gzxHtVoiceTextService;

    private final QaKeywordInitializationConfig qaKeywordInitializationConfig;

    @Value("${ai.deepseekR1.modelId}")
    private String modelId;

    @Value("${ai.deepseekR1.Bearer}")
    private String bearer;

    // 使用固定线程池 + 速率限制
    private static final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
    // 每秒多少次请求
    private static final RateLimiter rateLimiter = RateLimiter.create(8);

    // 最大重试次数
    private static final int MAX_RETRIES = 1;
    // 重试延迟
    private static final long RETRY_DELAY_MS = 1000;

    //使用AI来判定是否违规
    @DS("qc")
    @Override
    public void aiScanner() {
        LambdaQueryWrapper<GzxVoiceCheck> wrapper = new LambdaQueryWrapper<>();
        wrapper.in(GzxVoiceCheck::getStatus, 1);
        //这里默认每次处理5条数据 线上质检的的定时任务是10秒查询一次，这里要改成每次查询5条，如果查太多了会频繁GC
        wrapper.last("LIMIT 5");
        List<GzxVoiceCheck> gzxVoiceChecks = gzxVoiceCheckService.selectList(wrapper);

        //质检关键字
        Set<String> keyWord = qaKeywordInitializationConfig.loadKeyWord();

        //分批处理
        List<List<GzxVoiceCheck>> batches = Lists.partition(gzxVoiceChecks, 5);
        for (List<GzxVoiceCheck> batch : batches) {
            CountDownLatch batchLatch = new CountDownLatch(batch.size());
            for (GzxVoiceCheck gzxVoiceCheck : batch) {
                executor.submit(() -> {
                            Long begin = System.currentTimeMillis();
                            int retryCount = 0;
                            while (retryCount <= MAX_RETRIES) {
                                //发给ai的消息
                                StringBuilder message = new StringBuilder();
                                // 检查每个文本是否包含关键字
                                StringBuilder keywordMatches = new StringBuilder();
                                try {
                                    // 阻塞直到获得许可
                                    rateLimiter.acquire();
                                    String callId = gzxVoiceCheck.getCallId();
                                    //质检过的状态
                                    gzxVoiceCheck.setStatus(2);
                                    List<GzxHtVoiceText> gzxHtVoiceTexts = gzxHtVoiceTextService.getTextByCallId(callId);

                                    for (GzxHtVoiceText text : gzxHtVoiceTexts) {
                                        String role = text.getSpeakerRole() == 2 ? "调解员" : "用户";
                                        message.append(role).append("说：").append(text.getContent()).append("。");

                                        //contentKeySet 说话内容
                                        for (String keyword : keyWord) {
                                            if (text.getContent().contains(keyword)) {
                                                keywordMatches.append(keyword).append("、");
                                            }
                                        }
                                    }
                                    log.info("callId: {} - 质检匹配到的关键字: {}", gzxVoiceCheck.getCallId(), keywordMatches);

                                    message.append("请告诉我，这段对话在催收场景下是否违规，比如坐席态度不好，骂人或者用户有投诉、情绪激动、自杀等危险信号，你只需要回复1，2，3，4。1代表客户违规，2代表坐席违规，3代表双方都违规，4代表双方都不违规。强调一下你只需要回复一个数字就行");
                                    List<String> response = AiScannerApi.callRemoteApi(message.toString(), modelId, bearer);
                                    //ai质检这里有时候接口不通，结果是没有给质检过的数据打标记，会造成每次质检的数据都是重复的
                                    if (response.size() > 1) {
                                        if (!keywordMatches.isEmpty()) {
                                            //命中了关键字
                                            gzxVoiceCheck.setKeyword(keywordMatches.toString());
                                            gzxVoiceCheck.setAiApproveStatus("1");
                                        } else {
                                            //没命中
                                            gzxVoiceCheck.setAiApproveStatus("4");
                                        }

                                        String reason = response.get(1);
                                        gzxVoiceCheck.setAiReasonContent(reason);
                                        gzxVoiceCheckService.updateById(gzxVoiceCheck);
                                    } else {
                                        //ai质检失败（ai也是质检过一遍了，得给他一个状态） 没命中
                                        gzxVoiceCheck.setAiApproveStatus("4");
                                        gzxVoiceCheckService.updateById(gzxVoiceCheck);
                                    }
                                    // 成功则退出循环
                                    break;
                                } catch (Exception e) {
                                    retryCount++;
                                    if (retryCount > MAX_RETRIES) {
                                        log.error("API调用达到最大重试次数 ", e);
                                        break;
                                    }
                                    log.error("API服务器过载，第{}次重试 ", retryCount);
                                    try {
                                        Thread.sleep(RETRY_DELAY_MS);
                                    } catch (InterruptedException ex) {
                                        log.error("线程被中断： 重试使用AI来判定是否违规 ", ex);
                                    }
                                } finally {
                                    Long end = System.currentTimeMillis();
                                    long time = end - begin;
                                    log.info("本次调用接口话费时间为 {} 毫秒", time);
                                    //ai没响应或者接口不通会用关键字表的数据代替ai质检
                                    if (!keywordMatches.isEmpty()) {
                                        //命中了关键字
                                        gzxVoiceCheck.setKeyword(keywordMatches.toString());
                                        gzxVoiceCheck.setAiApproveStatus("1");
                                    } else {
                                        //没命中
                                        gzxVoiceCheck.setAiApproveStatus("4");
                                    }
                                    gzxVoiceCheckService.updateById(gzxVoiceCheck);
                                }
                            }
                        }
                );
            }
            try {
                // 等待当前批次完成
                batchLatch.await();
                // 批次间隔
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                log.error("线程被中断： 使用AI来判定是否违规 ", e);
            }
        }

    }
}
